package pkg

import (
	"context"
	"sync"
	"time"

	"github.com/panjf2000/ants/v2"
)

type GlobalCType chan interface{}

type GeneralControl struct {
	InnerData chan interface{}
	// 消费globalDataCh的数据
	consumer         func(GlobalCType, GeneralControl)
	ErrCh            chan interface{}
	ctxTimOut        contextTimeOut
	errStop          chan struct{}
	innerDataChOnce  *sync.Once
	globalDataChOnce *sync.Once
	globalDataCh     GlobalCType
	completeCh       chan struct{}

	stopMsgCh chan struct{}
}

type contextTimeOut struct {
	// 超时上下文
	timOutCtx         context.Context
	timeOutCancleFunc context.CancelFunc
}

func handleAction(config GeneralControl, globalCh GlobalCType) {
	handleDataFunc := config.consumer
	go handleDataFunc(globalCh, config)
	for {
		select {
		case <-config.ErrCh:
			goto ForErrEnd
		case <-config.completeCh:
			goto ForCompleteEnd
		case <-config.stopMsgCh:
			goto ForCompleteEnd
		case <-config.ctxTimOut.timOutCtx.Done():
			goto ForTimeOutEnd
		}

	}
ForTimeOutEnd:
	return
ForErrEnd:
	return
ForCompleteEnd:
	return
}

// SendMsg  goroutineCount 协程池大小,timeOutDuration 超时时间设置,importDataFunc 数据带入方法
func SendMsg(goroutineCount int, timeOutDuration time.Duration, importDataFunc func(innerDataCh, errCh chan<- interface{}, globalDataCh GlobalCType)) {
	runTimes := goroutineCount
	ctxTimeOut, cancleFunTimeOut := context.WithDeadline(context.Background(), time.Now().Add(timeOutDuration))
	defer ants.Release()
	defer cancleFunTimeOut()

	generalConfig := GeneralControl{}
	generalConfig.globalDataCh = make(GlobalCType, goroutineCount)
	generalConfig.ctxTimOut.timOutCtx = ctxTimeOut
	generalConfig.ctxTimOut.timeOutCancleFunc = cancleFunTimeOut
	generalConfig.InnerData = make(chan interface{})
	generalConfig.ErrCh = make(chan interface{})
	generalConfig.stopMsgCh = make(chan struct{})
	generalConfig.innerDataChOnce = &sync.Once{}
	generalConfig.globalDataChOnce = &sync.Once{}
	generalConfig.completeCh = make(chan struct{})
	generalConfig.consumer = receiveFunc
	jk := func() {
		importDataFunc(generalConfig.InnerData, generalConfig.ErrCh, generalConfig.globalDataCh)
		generalConfig.globalDataChOnce.Do(func() {
			close(generalConfig.globalDataCh)
		})
	}
	go jk()
	go Result(generalConfig.InnerData, generalConfig.ErrCh, generalConfig.stopMsgCh)
	var wg sync.WaitGroup
	p, _ := ants.NewPoolWithFunc(runTimes, func(i interface{}) {
		handleAction(generalConfig, generalConfig.globalDataCh)
		wg.Done()
	}, func(opts *ants.Options) {
		// opts.ExpiryDuration = time.Duration(10) * time.Second
	})
	defer p.Release()
	for i := 0; i < runTimes; i++ {
		wg.Add(1)
		_ = p.Invoke(generalConfig)
	}
	wg.Wait()

}

func receiveFunc(globalData GlobalCType, generalConfig GeneralControl) {
	for data := range globalData {
		HandleMsg(data, generalConfig.InnerData, generalConfig.ErrCh)
	}
	generalConfig.completeCh <- struct{}{}
}

type HandleMsgFunc func(data interface{}, innerData, err chan<- interface{})

var HandleMsg HandleMsgFunc

type ResultFunc func(innerDataCh, errCh <-chan interface{}, stopMsgCh chan<- struct{}) (innerData, err []interface{})

var Result ResultFunc
